package com.zhike.utils;

import com.zhike.bean.TransientSink;
import com.zhike.constant.DorisConstant;
import lombok.SneakyThrows;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.SimpleStringSerializer;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Properties;

/**
 * @Author: zhike
 * @CreateTime: 2024/1/26
 * @Description:   创建DorisSink
 */
public class DorisUtils {
    public static DorisSink<String> getDorisSink(String table){
        Properties properties = new Properties();
        properties.setProperty("format", "json");
        properties.setProperty("read_json_by_line", "true");

        return DorisSink.<String>builder()  // 创建一个DorisSink对象用于将数据写入Doris数据库
                .setDorisOptions(  // 设置Doris数据库的连接选项
                        DorisOptions.builder()  // 创建一个DorisOptions对象
                                .setFenodes(DorisConstant.HOSTNAME + ":" + DorisConstant.PORT)  // 设置Fenode服务器地址
                                .setTableIdentifier(DorisConstant.DORIS_SCHEMA + "." + table)  // 设置表的标识符
                                .setUsername(DorisConstant.USERNAME)  // 设置用户名
                                .setPassword(DorisConstant.PASSWORD)  // 设置密码
                                .build()  // 构建DorisOptions对象
                )
                .setDorisExecutionOptions(  // 设置Doris数据库的执行选项
                        DorisExecutionOptions.builder()  // 创建一个DorisExecutionOptions对象
                                .setLabelPrefix("car"+System.currentTimeMillis())  // 设置标签前缀
                                .setDeletable(false)  // 设置是否可删除
                                .setStreamLoadProp(properties)  // 设置流加载属性
                                .build()  // 构建DorisExecutionOptions对象
                )
                .setDorisReadOptions(  // 设置Doris数据库的读取选项
                        DorisReadOptions.builder().build()  // 创建一个DorisReadOptions对象并构建
                ).setSerializer(new SimpleStringSerializer())  // 设置数据序列化方式为SimpleStringSerializer
                .build();
    }



    //通过JDBC方式将数据写入Doris
    public static <T> SinkFunction<T> getJdbcSink(String sql){
        return JdbcSink.<T>sink(
                sql,
                new JdbcStatementBuilder<T>() {
                    @SneakyThrows
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t)  {
                        Class<?> aClass = t.getClass();
                        Field[] fields = aClass.getDeclaredFields();

                        int offest = 0;
                        for (int i = 0; i < fields.length; i++) {
                            Field field = fields[i];
                            field.setAccessible(true);

                            //尝试获取该属性是否有注解
                            TransientSink annotation = field.getAnnotation(TransientSink.class);
                            if (annotation != null) {
                                offest++;
                                continue;
                            }
                            //获取字段的值
                            Object value = field.get(t);
                            //给占位符赋值
                            preparedStatement.setObject(i + 1, value);

                        }
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchIntervalMs(2000L)
                        .withBatchSize(5)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(DorisConstant.JDBC_DRIVER)
                        .withUrl(DorisConstant.JDBC_URL)
                        .withUsername("root")
                        .withPassword("123456")
                        .build()
        );


    }

}
